package cas.ihep.hadoop.util;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hdfs.DFSClient;
import org.apache.hadoop.util.Time;

import java.io.Closeable;
import java.io.File;
import java.io.IOException;
import java.net.SocketTimeoutException;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Future;

public class RefreshFileSize implements Closeable{

    private static final Log LOG = LogFactory.getLog(RefreshFileSize.class);
    private static final long RENEWER_SLEEP_DEFAULT = 500L;
    private static final long RENEWER_INTERVAL_DEFAULT = 5000L;

    private static class Singleton{
        private static final RefreshFileSize instance;
        static{
            instance=new RefreshFileSize();
            Runtime.getRuntime().addShutdownHook(new Thread(){
                public void run(){
                    instance.close();
                }
            });
        }
    }

    private RefreshFileSize() {
    }

    public static RefreshFileSize instance(){
        return Singleton.instance;
    }

    //private int currentId = 0;
    //private Daemon daemon = null;
    private static class Tuple{
        long fileId;
        DFSClient dfsc;
        File file;
        Tuple(long l,String p,DFSClient c){
            fileId=l;
            dfsc=c;
            file=new File(p);
        }
        public boolean equals(Object obj){
            if(obj instanceof Tuple){
                Tuple t=(Tuple)obj;
                return fileId==t.fileId;
            }
            return false;
        }
    }

    private final ConcurrentHashMap<String,Tuple> files = new ConcurrentHashMap<>();

    private class InternalCallable implements Callable<Integer> {
        volatile boolean stop;
        InternalCallable(){
            stop=false;
        }
        @Override
        public Integer call() {
            try {
                long last=0;
                for (;!stop;) {
                    long now=Time.now();
                    if (now-last >= RENEWER_INTERVAL_DEFAULT) {
                        try {
                            renew();
                            last = now;
                        } catch (SocketTimeoutException ie) {
                            break;
                        } catch (IOException ie) {
                            LOG.warn(ie.getMessage());
                        }
                    }
                    Thread.sleep(RENEWER_SLEEP_DEFAULT);
                }
            }catch (InterruptedException e){
                LOG.error("ERROR in daemon: "+e.getMessage());
            }
            return 0;
        }
    }

    private InternalCallable iDaemon=null;
    private Future<Integer> iFuture=null;

    public void put(String src,long fileId,String path,DFSClient dfsc){
        if(iDaemon==null){
            synchronized (this){
                if(iDaemon==null){
                    iDaemon=new InternalCallable();
                    iFuture=YbjThreadPool.pool.submit(iDaemon);
                }
            }
        }
        files.putIfAbsent(src,new Tuple(fileId,path,dfsc));
        LOG.info("Updating "+files.size()+" files\' size");
    }

//    synchronized public void put(final String src, final long fileId, final String path) {
//        if (!isRunning()) {
//            //start a new deamon with a new id.
//            final int id = ++currentId;
//            daemon = new Daemon(new Runnable() {
//                @Override
//                public void run() {
//                    try {
//                        RefreshFileSize.this.run(id);
//                    } catch(InterruptedException e) {
//                        LOG.info("ERROR:"+e.getMessage());
//                    }
//                }});
//            daemon.start();
//        }
//        addFile(src,fileId,path);
//    }

    /**
     private synchronized long getSleepPeriod() {
     long sleepPeriod = RENEWER_SLEEP_DEFAULT;
     return sleepPeriod;
     }
     */
    /** @return the renewal time in milliseconds. */
    /**
     private synchronized long getRenewalTime() {
     long renewal = RENEWER_INTERVAL_DEFAULT;
     return renewal;
     }
     */
    /** Is the daemon running? */
//    private synchronized boolean isRunning() {
//        return daemon != null && daemon.isAlive();
//    }
//    /** Add a file. */
//    private synchronized void addFile(final String src, final long fileId, final String path) {
//        if(files.containsKey(src)) {
//            return;
//        }
//        Map<Long,String> map = new HashMap<>();
//        map.put(fileId,path);
//        files.put(src,map);
//        LOG.info("files size : "+files.size());
//    }
    /**
     * Periodically check in with the namenode and renew all the leases
     * when the lease period is half over.
     */
//    private void run(final int id) throws InterruptedException {
//        for(long lastRenewed = Time.now(); !Thread.interrupted();
//            Thread.sleep(RENEWER_SLEEP_DEFAULT)) {
//            final long elapsed = Time.now() - lastRenewed;
//            if (elapsed >= RENEWER_INTERVAL_DEFAULT) {
//                try {
//                    renew();
//                    lastRenewed = Time.now();
//                } catch (SocketTimeoutException ie) {
//                    break;
//                } catch (IOException ie) {
//                    LOG.warn(ie.getMessage());
//                }
//            }
//            synchronized(this) {
//                if (id != currentId ) {
//                    //no longer the current daemon or expired
//                    return;
//                }
//            }
//        }
//    }

//    private void renew() throws IOException {
//        final Map<String,Map<Long,String>> copies;
//        synchronized(this) {
//            copies = new HashMap<>(files);
//        }
//        String previousFile = null;
//        for (String src : copies.keySet()) {
//            if (!src.equals(previousFile)) {
//                for(long id : copies.get(src).keySet()){
//                    String path = copies.get(src).get(id);
//                    File file = new File(path);
//                    if(file.exists()){
//                        long size = file.length();
//                        dfsc.getNamenode().fsync(src, id, dfsc.getClientName(), size);
//                    }
//                }
//
//            }
//            previousFile = src;
//        }
//    }

    private void renew()throws IOException{
        String previousFile=null;
        for(Map.Entry<String,Tuple> srcEntry:files.entrySet()){
            String src=srcEntry.getKey();
            if(!src.equals(previousFile)){
                Tuple t=srcEntry.getValue();
                if(t.file.exists()){
                    t.dfsc.getNamenode().fsync(src,t.fileId,t.dfsc.getClientName(),t.file.length());
                }
            }
            previousFile=src;
        }
    }
//    public synchronized void remove(String src) {
//        files.remove(src);
//        int sz=files.size();
//        LOG.info("files size : "+sz);
//        if(sz < 1 && daemon.isAlive()){
//            daemon.interrupt();
//        }
//    }
//    public void close(){
//        files.clear();
//        if(daemon.isAlive()){
//            daemon.interrupt();
//        }
//    }

    public void remove(String src){
        files.remove(src);
        if(files.isEmpty()){
            if(iDaemon!=null){
                synchronized (this){
                    if(iDaemon!=null){
                        iDaemon.stop=true;
                        try{
                            iFuture.get();
                        }catch (Exception ignored){}
                        iDaemon=null;
                        iFuture=null;
                    }
                }
            }
        }
    }

    public void close(){
        if(iDaemon!=null){
            synchronized (this){
                if(iDaemon!=null){
                    iDaemon.stop=true;
                    try{
                        iFuture.get();
                    }catch (Exception ignored){}
                    iDaemon=null;
                    iFuture=null;
                }
            }
        }
        files.clear();
    }
}


